package com.sbwebsocket.core.redis;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sbwebsocket.core.model.bo.UnusualMessage;
import com.sbwebsocket.core.redis.model.contants.WsMessageRedisKeyPrefix;
import com.sbwebsocket.core.service.UnusualMessageManager;
import lombok.Getter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 周期性循环处理异常掉线的客户端
 * @Auther: GSJ
 * @Date: 2022/2/16
 * @Description:
 * @version: 1.0
 */
//@Component
public class CycleProcessUnusualOfflineClientMessage {

    private final DelayQueue<DelayMessage<String>> delayQueue;
    ThreadPoolExecutor threadPoolExecutor;
    @Autowired
    UnusualMessageManager redisUnusualOfflineClientMessageManager;
    @Autowired
    StringRedisTemplate redisTemplate;

    public CycleProcessUnusualOfflineClientMessage() {
        this.delayQueue = new DelayQueue();
        threadPoolExecutor = new ThreadPoolExecutor(1,
                Runtime.getRuntime().availableProcessors() * 2,
                1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread();
                        thread.setName("CycleUOCMessageProcess");
                        return thread;
                    }
                }, new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 处理异常的消息
     */
    public void processUnsualClientMsg() {

        //todo: 可用接口取异常(延迟)消息，这样消息存取可自行实现
        Set<ZSetOperations.TypedTuple<String>> unusual =
                redisTemplate.opsForZSet().rangeByScoreWithScores(WsMessageRedisKeyPrefix.UNUSUAL_MSG_LIST, 0,
                System.currentTimeMillis(), 0, 1);
        for (ZSetOperations.TypedTuple<String> stringTypedTuple : unusual) {
            Double score = stringTypedTuple.getScore();
            String value = stringTypedTuple.getValue();

            delayQueue.put(new DelayMessage(score.longValue(), value));
            redisTemplate.opsForZSet().remove(WsMessageRedisKeyPrefix.UNUSUAL_MSG_LIST,value);
        }
    }


    /**
     * 消费消息
     */
    @PostConstruct
    public void consumer(){
        new Thread(()->{
            while (true) {
                try {
                    DelayMessage<String> take = this.delayQueue.take();
//                    threadPoolExecutor.submit(() -> {
                        redisUnusualOfflineClientMessageManager.reput(take.getData());
//                    });
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public static void main(String[] args) throws JsonProcessingException {
        long l = 20000 + System.currentTimeMillis();
        System.out.println(l);
        UnusualMessage a = new UnusualMessage("2","test");
        ObjectMapper objectMapper = new ObjectMapper();
        String s = objectMapper.writeValueAsString(a);
        System.out.println(s);

    }

    @Getter
    public static class DelayMessage<T> implements Delayed {

        private Long activeTime;

        private T data;

        public DelayMessage(Long activeTime, T data) {
            this.activeTime = activeTime;
            this.data = data;
        }

        /**
         * 这个方法返回到激活日期的剩余时间，时间单位由单位参数指定。
         * @param unit
         * @return
         */
        @Override
        public long getDelay(TimeUnit unit) {
            long d = unit.convert(this.activeTime
                    - System.currentTimeMillis(), unit);
            return d;
        }

        /**
         *Delayed接口继承了Comparable接口，按剩余时间排序，实际计算考虑精度为纳秒数
         */
        @Override
        public int compareTo(Delayed o) {
            long d = (getDelay(TimeUnit.MILLISECONDS)
                    - o.getDelay(TimeUnit.MILLISECONDS));
            if (d == 0) {
                return 0;
            } else {
                if (d < 0) {
                    return -1;
                } else {
                    return 1;
                }
            }
        }
    }
}
